package com.itc.bi.service.impl;

import com.alibaba.fastjson2.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.google.common.collect.Lists;
import com.itc.bi.bo.DataSetColumnTypeBO;
import com.itc.bi.constant.RedisConstant;
import com.itc.bi.constant.SystemConstant;
import com.itc.bi.dto.*;
import com.itc.bi.entity.*;
import com.itc.bi.enums.ProcessEnum;
import com.itc.bi.mapper.DataProcessMapper;
import com.itc.bi.mapper.DataSetColumnMapper;
import com.itc.bi.mapper.DirectoryMapper;
import com.itc.bi.service.DataProcessInfoService;
import com.itc.bi.mapper.DataProcessInfoMapper;
import com.itc.bi.service.DataSetColumnService;
import com.itc.bi.service.FunctionConfigService;
import com.itc.bi.utils.Constants;
import com.itc.bi.utils.FilterComponent.Node;
import com.itc.bi.utils.FilterExp;
import com.itc.bi.utils.createTableUtils;
import com.itc.bi.vo.DataByDtIdVo;
import com.itc.common.core.exception.CheckedException;
import com.itc.common.core.exception.ServiceException;
import com.itc.common.security.utils.SecurityUtils;
import lombok.extern.log4j.Log4j2;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author hecai
 * @description 针对表【tb_data_process_info(数据加工详情)】的数据库操作Service实现
 * @createDate 2024-06-27 18:35:37
 */
@Service
@Log4j2
public class DataProcessInfoServiceImpl extends ServiceImpl<DataProcessInfoMapper, DataProcessInfoEntity> implements DataProcessInfoService {

    @Autowired
    private ClickHouseJDBCService clickHouseJDBCService;
    @Autowired
    DataProcessMapper processMapper;
    @Autowired
    DirectoryMapper directoryMapper;
    @Autowired
    RedissonClient redissonClient;
    @Autowired
    DataSetColumnService dataSetColumnService;
    @Autowired
    DataSetColumnMapper dataSetColumnMapper;
    @Autowired
    FunctionConfigService functionConfigService;
    private final String BEFORE = "before";
    private final String AFTER = "after";

    @Override
    public List exec(DataProcessInfoDto dto) {
        DataProcessInfoEntity info = new DataProcessInfoEntity();
        DataProcessInfoEntity entity = null;
        if(null != dto.getId()){
            entity = baseMapper.selectById(dto.getId());
            if(null == entity){
                throw new ServiceException("记录不存在");
            }
            info.setId(dto.getId());
            info.setOrderNum(entity.getOrderNum());
        }else{
            entity = super.baseMapper.getLatestByProcessId(dto.getDpId());
            if (null != entity) {
                info.setOrderNum(entity.getOrderNum() + 1);
            } else {
                info.setOrderNum(1);
            }
        }

        info.setDpId(dto.getDpId());
        info.setOriginalParameter(dto.getParam());
        info.setType(dto.getType());
        init(info);
        String sql = "select * from (" + info.getParameterValues() + ") limit 0,50";
        return clickHouseJDBCService.queryTableBySql(sql);
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Long saveInfo(DataProcessInfoDto dto) {
        DataProcessInfoEntity info = new DataProcessInfoEntity();
        DataProcessInfoEntity old = null;

        int saveType = 1;//1:新增数据加工新增  2:新增数据加工编辑 3:编辑数据加工新增  4:编辑数据加工编辑

        DataProcessEntity process = processMapper.selectById(dto.getDpId());
        if(null != process){
            saveType= 3;
        }

        //生成数仓表名
        String table = createTableUtils.createTableName(Constants.PROCESS_DATA, dto.getType());
        if(null != dto.getId()){
            old = baseMapper.selectById(dto.getId());
            if(null == old){
                throw new ServiceException("记录不存在");
            }
            if(!old.getType().equals(dto.getType())){
                throw new ServiceException("加工类型不可变更");
            }
            saveType++;
            info.setId(dto.getId());
            info.setOrderNum(old.getOrderNum());
        }else{
            old = super.baseMapper.getLatestByProcessId(dto.getDpId());
            if (null != old) {
                info.setOrderNum(old.getOrderNum() + 1);
            } else {
                info.setOrderNum(1);
            }
        }

        info.setDpId(dto.getDpId());
        info.setOriginalParameter(dto.getParam());
        info.setProcessTable(table);
        info.setType(dto.getType());

        //初始化差异性字段值
        init(info);

        if (StringUtils.isNotBlank(info.getColumnJson())) {
            if(saveType%2==0){//编辑
                //原始加工数据产生的字段
                List<DataSetColumnEntity> oldColumnList = JSON.parseArray(old.getColumnJson(), DataSetColumnEntity.class);
                //编辑后加工数据产生的字段
                List<DataSetColumnEntity> columnList = JSON.parseArray(info.getColumnJson(), DataSetColumnEntity.class);
                Set<String> oldColumnNameSet = oldColumnList.stream().map(DataSetColumnEntity::getColumnName).collect(Collectors.toSet());
                oldColumnNameSet.removeAll(columnList.stream().map(DataSetColumnEntity::getColumnName).collect(Collectors.toSet()));
                //字段有变化是校验变化的字段是否被使用
                if(oldColumnNameSet.size() > 0){
                    Set<String> set = new HashSet<>(); //删除记录步骤后的加工步骤使用到的字段,用来判断删除数据增加的字段是否被后续使用
                    LambdaQueryWrapper<DataProcessInfoEntity> qw = Wrappers.lambdaQuery();
                    qw.eq(DataProcessInfoEntity::getDpId, info.getDpId());
                    qw.gt(DataProcessInfoEntity::getOrderNum,info.getOrderNum());
                    List<DataProcessInfoEntity> list = list(qw);
                    for (DataProcessInfoEntity item : list){
                        if (StringUtils.isNotBlank(item.getUserField())) {
                            set.addAll(Arrays.stream(item.getUserField().split("、")).collect(Collectors.toSet()));
                        }
                    }
                    set.retainAll(oldColumnNameSet);
                    if (set.size()>0) {
                        throw new ServiceException(String.format("字段 %s 被使用", set.toString()));
                    }
                }

                //查询数据加工所有字段
                LambdaQueryWrapper<DataSetColumnEntity> qw = Wrappers.lambdaQuery();
                qw.eq(DataSetColumnEntity::getDataSetId,dto.getDpId());
                qw.eq(DataSetColumnEntity::getStepNum,info.getOrderNum());

                List<DataSetColumnEntity> infoList = dataSetColumnService.list(qw);

                //字段数量没有变化
                if(oldColumnList.size() == columnList.size()){
                    for (int i = 0; i < infoList.size(); i++) {
                        DataSetColumnEntity oldcolumn = infoList.get(i);
                        DataSetColumnEntity column = columnList.get(i);
                        oldcolumn.setColumnName(column.getColumnName());
                        oldcolumn.setColumnComment(column.getColumnComment());
                        oldcolumn.setColumnType(column.getColumnType());
                    }
                    dataSetColumnService.updateBatchById(infoList);
                }else{//字段数发生变化，将当前步骤字段加上后续步骤的字段信息初始化对应的位置信息
                    dataSetColumnService.removeBatchByIds(infoList);
                    List<DataSetColumnEntity> updateList = new ArrayList<>();
                    columnList.stream().forEach(item->{
                        item.setDataSetId(dto.getDpId());
                        item.setStepNum(info.getOrderNum());
                    });
                    updateList.addAll(columnList);
                    updateList.addAll(stepColumnList(info,AFTER));

                    List<DataSetColumnEntity> beforeList = stepColumnList(info,BEFORE);
                    int minPosition = beforeList.size();
                    for (DataSetColumnEntity column : updateList) {
                        column.setColumnPosition(minPosition);
                        minPosition++;
                    }
                    dataSetColumnService.saveOrUpdateBatch(updateList);
                }
            }else{//新增
                List<DataSetColumnEntity> columnList = JSON.parseArray(info.getColumnJson(), DataSetColumnEntity.class);
                LambdaQueryWrapper<DataSetColumnEntity> columnWq = Wrappers.lambdaQuery();
                columnWq.eq(DataSetColumnEntity::getDataSetId, dto.getDpId());
                List<String> columnNames = columnList.stream().map(DataSetColumnEntity::getColumnName).collect(Collectors.toList());
                columnWq.in(DataSetColumnEntity::getColumnName, columnNames);
                List<DataSetColumnEntity> list = dataSetColumnService.list(columnWq);
                if (!CollectionUtils.isEmpty(list)) {
                    String columnName = "";
                    for (DataSetColumnEntity item : list) {
                        columnName = columnName.concat(item.getColumnName()).concat("、");
                    }
                    throw new ServiceException(String.format("字段%s已经存在", columnName.substring(0, columnName.length() - 1)));
                }
                for (DataSetColumnEntity item : columnList){
                    item.setStepNum(info.getOrderNum());
                }
                dataSetColumnService.insertList(columnList);
            }
            dataSetColumnService.redisCache(dto.getDpId());
        }
        //数仓创建表
        clickHouseJDBCService.createTableProcessDS(info.getProcessTable(), dto.getDpId(), info.getParameterValues());

        if(ProcessEnum.UNION.getType().equals(info.getType())){
            info.setParameterValues(info.getParameterValues().replace(info.getUnionTableName(),"${unionTableName}"));
        }

        saveOrUpdate(info);

        //编辑时删除原表
        if(saveType%2==0){
            afterStep(info,0);
            clickHouseJDBCService.dropTable(old.getProcessTable());
        }
        editDear(info,process);

        return info.getId();
    }

    /**
     * 根据加工ID获取数仓中的表名,当orderNum为null是获取最新一条,不为null时去orderNum的前一条
     *
     * @param dpId
     * @param orderNum
     * @return
     */
    public String getTableName(Long dpId, Integer orderNum) {
        DataProcessInfoEntity entity = null;
        if (null == orderNum) {
            entity = super.baseMapper.getLatestByProcessId(dpId);
        } else {
            entity = super.baseMapper.getPreviousData(dpId, orderNum);
        }
        if (null != entity) {
            return entity.getProcessTable();
        } else {
            String redisKey = RedisConstant.DATA_SET_PROCESS + SecurityUtils.getUserId() + ":";
            Long dtId = (Long) redissonClient.getBucket(redisKey + dpId).get();
            if (null == dtId) {
                DataProcessEntity process = processMapper.selectById(dpId);
                if(null == process){
                    throw new ServiceException("加工数据不存在");
                }
                dtId = process.getDtId();
            }
            DirectoryEntity directory = directoryMapper.selectById(dtId);
            if (null == directory) {
                throw new ServiceException("数据集不存在");
            }
            return directory.getTableName();
        }
    }

    private void init(DataProcessInfoEntity info) {
        String type = info.getType();
        Long dpId = info.getDpId();

        ProcessEnum processEnum = ProcessEnum.get(type);
        if (null == processEnum) {
            throw new CheckedException("类型不支持");
        }
        info.setName(processEnum.getName());

        String tableName = getTableName(dpId, info.getOrderNum());
        String jsonParam = info.getOriginalParameter();
        List<DataSetColumnEntity> columnList = new ArrayList<>();
        switch (processEnum) {
            case FILTER:
                Node node = JSON.parseObject(jsonParam, Node.class);
                //获取过滤表达式
                FilterExp exp = new FilterExp(node);
                String filterSql = "select "+dpId+" `"+SystemConstant.DATABASEREDUNDANCYCOLUMN +"`,"+ stepBeforeColumns(info,null) + " from " + tableName + " where " + exp.exp();
                info.setParameterValues(filterSql);
                info.setUserField(exp.getFields());
                return;
            case CALCULATION:
                //设置新字段属性
                CalculationDto calculationDto = JSON.parseObject(jsonParam, CalculationDto.class);
                DataSetColumnEntity column = new DataSetColumnEntity();
                column.setColumnName(calculationDto.getColumnName());
                column.setColumnComment(calculationDto.getColumnComment());
                column.setDataSetId(dpId);

                functionConfigEntity function = null;
                if(null != calculationDto.getFunctionId()){
                    function = functionConfigService.getById(calculationDto.getFunctionId());
                }

                //查询新字段数据类型
                String querySql = "select (" + calculationDto.getExpression() + ") `" + calculationDto.getColumnName() + "` from " + tableName + " limit 0,1";
                List<DataSetColumnTypeBO> caDatalist = clickHouseJDBCService.getColumnListBySql(querySql);
                DataSetColumnTypeBO bo= caDatalist.stream().filter(v -> v.getColumnName().equals(column.getColumnName())).findFirst().orElse(null);
                if(null == bo){
                    column.setColumnType(SystemConstant.DATA_TYPE_STRING);
                }else{
                    column.setColumnType(bo.getColumnType());
                }
                columnList.add(column);

                StringBuilder calBuffer = new StringBuilder("select ");
                calBuffer.append(dpId).append(" ").append(SystemConstant.DATABASEREDUNDANCYCOLUMN)
                        .append(",").append(stepBeforeColumns(info,null)).append(",(");
                if(null == function || !"union".equals(function.getFunctionType())){
                    //数字保留2位小数
                    if(SystemConstant.DATA_TYPE_NUMBER.equals(column.getColumnType())){
                        calBuffer.append("round(").append(calculationDto.getExpression()).append(",2)");
                    }else{
                        calBuffer.append(calculationDto.getExpression());
                    }
                }else{
                    //数字保留2位小数
                    if(SystemConstant.DATA_TYPE_NUMBER.equals(column.getColumnType())){
                        calBuffer.append(" select round(").append(calculationDto.getExpression()).append(",2) ");
                    }else{
                        calBuffer.append(" select ").append(calculationDto.getExpression());
                    }
                }
                calBuffer.append(") `").append(calculationDto.getColumnName()).append("` from ").append(tableName);

                info.setParameterValues(calBuffer.toString());

                info.setColumnJson(JSON.toJSONString(columnList));

                List<String> expColumns = Arrays.asList(calculationDto.getExpression().split("`"));

                List<DataSetColumnEntity> stepColumns = stepColumnList(info, BEFORE);
                List<String> useColumn = stepColumns.stream().map(item -> item.getColumnName()).filter(item -> expColumns.contains(item)).collect(Collectors.toList());

                info.setUserField(String.join("、",useColumn));
                return;
            case GROUP:
                GroupAssignParameterDto groupAssignParameterDto = JSON.parseObject(jsonParam, GroupAssignParameterDto.class);
                if (CollectionUtils.isEmpty(groupAssignParameterDto.getDataRange())) {
                    throw new SecurityException("区间不存在,请检查");
                }
                String caseColumnName = groupAssignParameterDto.getAssignBasis();
                DataSetColumnEntity columnEntity = dataSetColumnService.getOne(Wrappers.lambdaQuery(DataSetColumnEntity.class).eq(DataSetColumnEntity::getDataSetId, dpId).eq(DataSetColumnEntity::getColumnName,caseColumnName));
                if(null == columnEntity){
                    throw new ServiceException(String.format("字段 %s 不存在",groupAssignParameterDto.getAssignBasis()));
                }
                String columnType = columnEntity.getColumnType();
                StringBuilder groupSql = new StringBuilder("SELECT ").append(dpId).append(" as ").append(SystemConstant.DATABASEREDUNDANCYCOLUMN);
                groupSql.append(",").append(stepBeforeColumns(info,null)).append(", case ");
                List<DataRangeDto> dataRange = groupAssignParameterDto.getDataRange().stream().sorted(Comparator.comparing(DataRangeDto::getSort)).collect(Collectors.toList());
                if(dataRange.size() == 1 && StringUtils.isBlank(dataRange.get(0).getMetricsValueLowerLimit())
                && StringUtils.isBlank(dataRange.get(0).getMetricsValueUpperLimit())){
                    throw new ServiceException("只有一个区间时上下限不能同时为空");
                }
                String rangeName = null;
                String lower = null;
                String upper = null;
                for (int i = 0; i < dataRange.size(); i++){
                    DataRangeDto item = dataRange.get(i);
                    rangeName = item.getRangeName();
                    lower = item.getMetricsValueLowerLimit();
                    upper = item.getMetricsValueUpperLimit();
                    Assert.hasText(rangeName,"区间名称不能为空");
                    if(i>0){
                        Assert.hasText(item.getMetricsValueLowerLimit(),"指标下限不能为空");
                    }
                    if(!SystemConstant.DATA_TYPE_NUMBER.equals(columnType)){
                        if(StringUtils.isNotBlank(lower)){
                            lower = "'"+lower+"'";
                        }
                        if(StringUtils.isNotBlank(upper)){
                            upper = "'"+upper+"'";
                        }
                    }
                    if(SystemConstant.DATA_TYPE_STRING.equals(columnType)){
                        lower.replace("、","','");
                        groupSql.append(" when ").append(columnNameDear(caseColumnName)).append(" in (")
                                .append(lower).append(") then '").append(item.getRangeName()).append("'");
                    }else{
                        groupSql.append( " when ");
                        if(StringUtils.isNotBlank(lower)){
                            groupSql.append(columnNameDear(caseColumnName)).append(">=").append(lower);
                            if(StringUtils.isNotBlank(upper)){
                                groupSql.append(" and ");
                            }
                        }
                        if(StringUtils.isNotBlank(upper)){
                            groupSql.append(columnNameDear(caseColumnName)).append("<").append(upper);
                        }
                        groupSql.append(" then '").append(rangeName).append("'");
                    }
                }
                groupSql.append(" else '' end as `").append(groupAssignParameterDto.getAssignRowName())
                        .append("` from ").append(tableName);

                //String groupAssignSql = splicingDataAssignRowSql(tableName, groupAssignParameterDto,info).toString();
                DataSetColumnEntity groupColumn = new DataSetColumnEntity();
                groupColumn.setColumnName(groupAssignParameterDto.getAssignRowName());
                groupColumn.setColumnComment(groupAssignParameterDto.getAssignRowDescription());
                groupColumn.setColumnType(SystemConstant.DATA_TYPE_STRING);
                groupColumn.setDataSetId(dpId);
                columnList.add(groupColumn);
                info.setColumnJson(JSON.toJSONString(columnList));
                info.setParameterValues(groupSql.toString());
                info.setDescription(groupAssignParameterDto.getAssignRowDescription());
                info.setUserField(groupAssignParameterDto.getAssignBasis());
                return;
            case SUM:
                DataCollectDto dataCollectDto = JSON.parseObject(jsonParam, DataCollectDto.class);
                String dataCollectSql = splicingDataCollectSql(tableName, dataCollectDto, info).toString();
                DataSetColumnEntity sumColumn = new DataSetColumnEntity();
                sumColumn.setColumnName(dataCollectDto.getCollectField());
                sumColumn.setColumnComment(dataCollectDto.getCollectFieldDescription());
                sumColumn.setColumnType(SystemConstant.DATA_TYPE_NUMBER);
                sumColumn.setDataSetId(dpId);
                columnList.add(sumColumn);
                info.setColumnJson(JSON.toJSONString(columnList));
                info.setParameterValues(dataCollectSql);
                info.setDescription(dataCollectDto.getCollectFieldDescription());
                Set set = new HashSet();
                set.add(dataCollectDto.getChooseGroupField());
                set.add(dataCollectDto.getChooseCollectField());
                info.setUserField(String.join("、",set));
                return;
            case UNION:
                DataSetUnionDto dataSetUnionDto = JSON.parseObject(jsonParam, DataSetUnionDto.class);
                Long needUnionDataSetId = dataSetUnionDto.getNeedUnionDataSetId();
                info.setUnionDtId(needUnionDataSetId);

                DirectoryEntity directoryEntity = directoryMapper.selectById(needUnionDataSetId);
                if (null == directoryEntity) {
                    throw new ServiceException("数据集不存在");
                }
                String unionTableName = directoryEntity.getTableName();
                info.setUnionTableName(unionTableName);

                if (dataSetUnionDto.getUnionType().equals(Constants.UNION_TYPE_TOP_BOTTOM)) {
                    List<UnionBasisDto> unionBasis = dataSetUnionDto.getUnionBasis();
                    if (!(unionBasis.size()>0)){
                        throw new ServiceException("合并依据不能为空！");
                    }
                    StringBuffer useField = new StringBuffer();
                    for(UnionBasisDto union : unionBasis){
                        useField.append(union.getCurrentBasis()).append("、");
                        union.setCurrentBasis(columnNameDear(union.getCurrentBasis()));
                        union.setUnionTableBasis(columnNameDear(union.getUnionTableBasis()));
                    }
                    useField.delete(0,1);

                    Map<String, String> sourceColumnToTargetColumnMap = unionBasis.stream().filter(item->StringUtils.isNoneBlank(item.getUnionTableBasis())).collect(Collectors.toMap(UnionBasisDto::getCurrentBasis, UnionBasisDto::getUnionTableBasis));

                    String columns = stepBeforeColumns(info,null);
                    StringBuilder sql = new StringBuilder("select * from (");
                    sql.append("select ").append(dpId).append(" as ").append(SystemConstant.DATABASEREDUNDANCYCOLUMN)
                            .append(",").append(columns).append(" from ").append(tableName)
                            .append(" union all select ").append(dpId).append(" as ").append(SystemConstant.DATABASEREDUNDANCYCOLUMN);

                    for (String item : columns.split(",")){
                        sql.append(",").append(sourceColumnToTargetColumnMap.getOrDefault(item,"null"));
                    }
                    sql.append(" from ").append(unionTableName).append(")");

                    info.setParameterValues(sql.toString());
                    info.setUserField(useField.toString());
                } else {
                    if(null == dataSetUnionDto.getUnionResultItem() || dataSetUnionDto.getUnionResultItem().size()==0){
                        throw new ServiceException("合并项不能为空");
                    }

                    //合并数据集字段
                    List<DataSetColumnEntity> dataSetColumnList = dataSetColumnService.list(new LambdaQueryWrapper<DataSetColumnEntity>()
                            .eq(DataSetColumnEntity::getDataSetId, dataSetUnionDto.getNeedUnionDataSetId()));
                    //数据加工字段
                    List<DataSetColumnEntity> dataProcessColumn = stepColumnList(info,BEFORE);

                    Map<String,DataSetColumnEntity> dpMap = dataProcessColumn.stream().collect(Collectors.toMap(DataSetColumnEntity::getColumnName, Function.identity()));
                    Map<String,DataSetColumnEntity> dtMap = dataSetColumnList.stream().collect(Collectors.toMap(DataSetColumnEntity::getColumnName, Function.identity()));

                    Map<String,String> columnNameToAliasNameMap = new HashMap<>();

                    //合并数据集字段集合
                    for (String item : dataSetUnionDto.getUnionResultItem()) {
                        DataSetColumnEntity dataSetColumnEntity = dtMap.get(item);
                        if (dataSetColumnEntity != null) {
                            DataSetColumnEntity uncionColumn = new DataSetColumnEntity();
                            uncionColumn.setColumnType(dataSetColumnEntity.getColumnType());
                            uncionColumn.setColumnComment(dataSetColumnEntity.getColumnComment());
                            if(dpMap.containsKey(item)){
                                uncionColumn.setColumnName(dataSetColumnEntity.getColumnName()+"1");
                            }else{
                                uncionColumn.setColumnName(dataSetColumnEntity.getColumnName());
                            }
                            uncionColumn.setDataSetId(dpId);
                            columnList.add(uncionColumn);
                            columnNameToAliasNameMap.put(item,uncionColumn.getColumnName());
                        }else{
                            throw new ServiceException("数据集不存在字段"+item);
                        }
                    }
                    String leftAndRightUnionSql = leftAndRightUnionSql(tableName, unionTableName,  dataSetUnionDto, columnNameToAliasNameMap,info).toString();
                    info.setParameterValues(leftAndRightUnionSql);

                    info.setColumnJson(JSON.toJSONString(columnList));
                    info.setUserField(String.join("、", dataSetUnionDto.getUnionBasis().stream().map(UnionBasisDto::getCurrentBasis).collect(Collectors.toSet())));
                }
                return;
            case POLYMERIC:
                return;
            default:
                return;
        }
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void removeById(Long id) {
        //获取记录
        DataProcessInfoEntity info = getById(id);
        if (null == info) {
            throw new CheckedException("记录不存在");
        }
        //编辑标志
        boolean editFlag = false;
        //数据加工被删除的不做处理
        DataProcessEntity process = processMapper.get(info.getDpId());
        if(null != process){
            editFlag = true;
            if(process.getIsDeleted().equals(1)){
                return;
            }
        }

        //根据加工ID获取加工详情列表
        LambdaQueryWrapper<DataProcessInfoEntity> qw = Wrappers.lambdaQuery();
        qw.eq(DataProcessInfoEntity::getDpId, info.getDpId());
        qw.orderByAsc(DataProcessInfoEntity::getOrderNum);
        List<DataProcessInfoEntity> list = list(qw);
        //编辑时最少保留一条加工数据
        if(editFlag && list.size()<2){
            throw new ServiceException("编辑时请至少保留一条加工数据");
        }
        int index = info.getOrderNum();//删除数据在加工步骤中的索引
        list = list.stream().filter(item->item.getOrderNum() > index).collect(Collectors.toList());

        int size = 0;
        //删除的数据有新字段，需要判断该字段在后面的加工步骤中是否使用到
        if (StringUtils.isNotBlank(info.getColumnJson())) {
            List<DataSetColumnEntity> columnList = JSON.parseArray(info.getColumnJson(), DataSetColumnEntity.class);
            size = columnList.size();
            Set<String> useSet = new HashSet<>(); //删除记录步骤后的加工步骤使用到的字段,用来判断删除数据增加的字段是否被后续使用
            for (DataProcessInfoEntity item : list){
                if (StringUtils.isNotBlank(item.getUserField())) {
                    useSet.addAll(Arrays.stream(item.getUserField().split("、")).collect(Collectors.toSet()));
                }
            }
            for (DataSetColumnEntity column : columnList) {
                if (useSet.contains(column.getColumnName())) {
                    throw new CheckedException(String.format("字段 %s 被使用", column.getColumnName()));
                }
            }

            LambdaQueryWrapper<DataSetColumnEntity> columnQw = Wrappers.lambdaQuery();
            columnQw.eq(DataSetColumnEntity::getDataSetId,info.getDpId());
            columnQw.eq(DataSetColumnEntity::getStepNum,info.getOrderNum());
            dataSetColumnService.remove(columnQw);
        }

        super.removeById(id);

        afterStep(info,1);
        editDear(info,process);

        List<DataSetColumnEntity> processColumnList =stepColumnList(info,AFTER);
        for(DataSetColumnEntity item : processColumnList){
            item.setColumnPosition(item.getColumnPosition()-size);
            item.setStepNum(item.getStepNum()-1);
        }
        dataSetColumnService.updateBatchById(processColumnList);
        dataSetColumnService.redisCache(info.getDpId());

        //删除数仓中的表
        clickHouseJDBCService.dropTable(info.getProcessTable());
    }

    /**
     * 拼接数据赋值列sql
     *
     * @return
     */
    public StringBuilder splicingDataAssignRowSql(String tableName, GroupAssignParameterDto groupAssignParameterDto, DataProcessInfoEntity info) {
        String assignBasis = groupAssignParameterDto.getAssignBasis();
        List<DataRangeDto> dataRange = groupAssignParameterDto.getDataRange();
        StringBuilder sql = new StringBuilder("SELECT ").append(info.getDpId()).append(" as ").append(SystemConstant.DATABASEREDUNDANCYCOLUMN);
        sql.append(",").append(stepBeforeColumns(info,null)).append(", ( CASE ");
        for (DataRangeDto dataRangeDto : dataRange) {
            sql.append(" WHEN ").append(dataRangeDto.getMetricsValueLowerLimit()).append(" <= ")
                    .append(columnNameDear(assignBasis)).append(" AND ").append(columnNameDear(assignBasis)).append(" < ")
                    .append(dataRangeDto.getMetricsValueUpperLimit()).append(" THEN '")
                    .append(dataRangeDto.getRangeName()).append("' ");
        }
        return sql.append(" ELSE '' END ) AS `").append(groupAssignParameterDto.getAssignRowName())
                .append("` FROM ").append(tableName);
    }

    /**
     * 拼接数据汇总ql
     *
     * @return
     */
    public StringBuilder splicingDataCollectSql(String tableName, DataCollectDto dataCollectDto, DataProcessInfoEntity info) {
        String chooseGroupFields = dataCollectDto.getChooseGroupField();
        String[] groupFieldArr = chooseGroupFields.split("、");
        chooseGroupFields = chooseGroupFields.replace("、","`,`");
        chooseGroupFields = "`"+chooseGroupFields+"`";
        StringBuilder sql = new StringBuilder("SELECT ");
        sql.append(info.getDpId()).append(" ").append(SystemConstant.DATABASEREDUNDANCYCOLUMN)
                .append(",").append(stepBeforeColumns(info,"a")).append(",b.`").append(dataCollectDto.getCollectField())
                .append("` FROM ").append(tableName).append(" AS a").append(" INNER JOIN ( SELECT ")
                .append(chooseGroupFields).append(", ROUND(").append(dataCollectDto.getChooseCollectMethod())
                .append("( `").append(dataCollectDto.getChooseCollectField()).append("` ), 2 ) AS `")
                .append(dataCollectDto.getCollectField()).append("` FROM ").append(tableName)
                .append(" GROUP BY ").append(chooseGroupFields).append(" ) AS b on ");
        int index = 0;
        for (String field : groupFieldArr){
            if(index > 0){
                sql.append(" and ");
            }
            sql.append("a.`").append(field).append("` = b.`").append(field).append("`");
            index++;
        }
        return  sql;
    }

    /**
     * 左右合并
     *
     * @param primaryTable
     * @param relevanceTable
     * @param dataSetUnionDto
     * @return
     */
    public StringBuilder leftAndRightUnionSql(String primaryTable, String relevanceTable, DataSetUnionDto dataSetUnionDto, Map<String,String> columnNameToAliasNameMap,DataProcessInfoEntity info) {
        StringBuilder sql = new StringBuilder("SELECT ");
        //合并字段
        StringBuilder unionSb = new StringBuilder(",");
        for (Map.Entry<String, String> entry : columnNameToAliasNameMap.entrySet()) {
            unionSb.append("b.`").append(entry.getKey()).append("`");
            if(!entry.getKey().equals(entry.getValue())){
                unionSb.append(" as `").append(entry.getValue()).append("`");
            }
            unionSb.append(",");
        }
        unionSb.deleteCharAt(unionSb.length() - 1);
        //合并条件
        StringBuilder condition = new StringBuilder("1=1");
        for (UnionBasisDto unionBasi : dataSetUnionDto.getUnionBasis()) {
            condition.append(" AND a.").append(columnNameDear(unionBasi.getCurrentBasis()))
                    .append(" = ").append("b.").append(columnNameDear(unionBasi.getUnionTableBasis()));
        }

        return sql.append(info.getDpId()).append(" ").append(SystemConstant.DATABASEREDUNDANCYCOLUMN)
                .append(",").append(stepBeforeColumns(info,"a")).append(unionSb).append(" FROM ")
                .append(primaryTable).append(" AS a")
                .append(dataSetUnionDto.getUnionType()
                        .equals(Constants.UNION_TYPE_LEFT) ? " LEFT JOIN " : " RIGHT JOIN ")
                .append(relevanceTable).append(" AS b")
                .append(" ON ").append(condition);

    }

    /**
     * 获取加工步骤之前的字段
     * @param info
     * @return
     */
    private String stepBeforeColumns(DataProcessInfoEntity info,String prefix) {
        LambdaQueryWrapper<DataSetColumnEntity> columnQw = Wrappers.lambdaQuery();
        columnQw.eq(DataSetColumnEntity::getDataSetId,info.getDpId());
        columnQw.lt(DataSetColumnEntity::getStepNum,info.getOrderNum());
        columnQw.orderByAsc(DataSetColumnEntity::getColumnPosition);
        List<DataSetColumnEntity> columnList = dataSetColumnService.list(columnQw);
        String columns = columnList.stream().map(item -> item.getColumnName()).collect(Collectors.joining("`,`"));
        columns ="`"+columns+"`";
        if(StringUtils.isNotBlank(prefix)){
            columns = prefix.concat(".").concat(columns);
            prefix = ","+prefix+".";
            columns = columns.replace(",",prefix);
        }
        return columns;
    }

    /**
     * 获取加工步骤之前的字段列表
     * @param info
     * @return
     */
    private List<DataSetColumnEntity> stepColumnList(DataProcessInfoEntity info,String type) {
        LambdaQueryWrapper<DataSetColumnEntity> columnQw = Wrappers.lambdaQuery();
        columnQw.eq(DataSetColumnEntity::getDataSetId,info.getDpId());
        columnQw.orderByAsc(DataSetColumnEntity::getColumnPosition);
        if("before".equals(type)){
            columnQw.lt(DataSetColumnEntity::getStepNum,info.getOrderNum());
        }else if("after".equals(type)){
            columnQw.gt(DataSetColumnEntity::getStepNum,info.getOrderNum());
        }

        List<DataSetColumnEntity> columnList = dataSetColumnService.list(columnQw);

        return columnList;
    }

    private String columnNameDear(String columnName){
        if(StringUtils.isBlank(columnName)){
            return null;
        }
        return "`"+columnName+"`";
    }

    /**
     * 数据加工变更后处理后续逻辑
     * @param info
     * @param subNum
     */
    private void afterStep(DataProcessInfoEntity info,int subNum){
        LambdaQueryWrapper<DataProcessInfoEntity> qw = Wrappers.lambdaQuery();
        qw.eq(DataProcessInfoEntity::getDpId, info.getDpId());
        qw.gt(DataProcessInfoEntity::getOrderNum,info.getOrderNum());
        qw.orderByAsc(DataProcessInfoEntity::getOrderNum);
        List<DataProcessInfoEntity> list = list(qw);
        if(list.size() > 0){
            for (DataProcessInfoEntity item : list) {
                //初始化后面步骤的sql,过滤只处理后面一条加工数据
                if(!ProcessEnum.FILTER.getType().equals(info.getType()) || item.getOrderNum() - info.getOrderNum() == 1){
                    try {
                        clickHouseJDBCService.dropTable(item.getProcessTable());
                    }catch (Exception e){}
                    init(item);
                    clickHouseJDBCService.createTableProcessDS(item.getProcessTable(), info.getDpId(), item.getParameterValues());
                    if(ProcessEnum.UNION.getType().equals(item.getType())){
                        item.setParameterValues(item.getParameterValues().replace(item.getUnionTableName(),"${unionTableName}"));
                    }
                }
                item.setOrderNum(item.getOrderNum()-subNum);
                updateById(item);
            }
        }
    }

    /**
     * 编辑时处理
     * @param info
     * @param process
     */
    private void editDear(DataProcessInfoEntity info,DataProcessEntity process){

        if(null != process){
            int index = info.getOrderNum();
            LambdaQueryWrapper<DataProcessInfoEntity> qw = Wrappers.lambdaQuery();
            qw.eq(DataProcessInfoEntity::getDpId, info.getDpId());
            qw.orderByAsc(DataProcessInfoEntity::getOrderNum);
            List<DataProcessInfoEntity> list = list(qw);
            if(1 == index){
                DirectoryEntity directory = directoryMapper.selectById(process.getDtId());
                DataProcessInfoEntity first = list.get(0);
                first.setParameterValues(first.getParameterValues().replace(directory.getTableName(),"${tableName}"));
                updateById(first);
            }
            if(index >= list.get(list.size()-1).getOrderNum()){
                DataProcessInfoEntity latest = list.get(list.size()-1);
                process.setTableName(latest.getProcessTable());
                process.setFullSql(latest.getParameterValues());
                processMapper.updateById(process);
            }
        }
    }

    @Override
    public void expressionsCheck(DataProcessExpressionsCheck check){
        Assert.hasText(check.getType(),"类型不能为空");
        Assert.notNull(check.getId(),"ID不能为空");
        Assert.hasText(check.getExpressions(),"公式不能为空");

        String tableName = null;

        if(ProcessEnum.POLYMERIC.getType().equals(check.getType())){
            DirectoryEntity dataSet = null;
            DataProcessEntity process = processMapper.selectById(check.getId());
            if(null != process){
                dataSet = directoryMapper.selectById(process.getDtId());
            }else{
                dataSet = directoryMapper.selectById(check.getId());
            }

            if(null == dataSet){
                throw new ServiceException("数据集不存在");
            }
            tableName = dataSet.getTableName();
        }else{
            Integer orderNum = null;
            if(null != check.getInfoId()){
                DataProcessInfoEntity info = getById(check.getInfoId());
                Assert.notNull(info,"加工详情不存在");
                orderNum = info.getOrderNum();
            }
            tableName = getTableName(check.getId(),orderNum);
        }

        String sql = "select "+check.getExpressions()+" as rs from "+tableName+" limit 1,100";

        clickHouseJDBCService.queryTableBySql(sql);

        if(ProcessEnum.POLYMERIC.getType().equals(check.getType())){
            List<DataSetColumnTypeBO> list = clickHouseJDBCService.getColumnListBySql(sql);
            if(CollectionUtils.isEmpty(list)){
                throw new ServiceException("获取数据类型异常");
            }
            if(!SystemConstant.DATA_TYPE_NUMBER.equals(list.get(0).getColumnType())){
                throw new ServiceException("公式计算结果只能为数值,当前计算结果类型为"+list.get(0).getColumnType());
            }
        }
    }
}




